package com.sjgs.gis.service;

import com.sjgs.common.Partitions;
import com.sjgs.common.utils.DirUtils;
import com.sjgs.common.utils.KryoUtils;
import com.sjgs.gis.domain.TileSet;
import com.sjgs.gis.domain.enumeration.TileSetType;
import com.sjgs.gis.fs.store.DefaultKVStore;
import com.sjgs.gis.errors.LocalFileIOException;
import com.sjgs.gis.errors.RocksIOException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.RocksDBException;
import org.rocksdb.WriteBatch;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

import static com.sjgs.common.Constants.*;


/**
 * 瓦片集存储服务
 *
 * @author jxw
 */
@Service
@ConditionalOnBean(DefaultKVStore.class)
public class TileSetService {

    private static final Log logger = LogFactory.getLog(TileSetService.class);

    private final DefaultKVStore DefaultKVStore;

    public TileSetService(DefaultKVStore DefaultKVStore) {
        this.DefaultKVStore = DefaultKVStore;
    }

    public Mono<TileSet> findTileSetById(String id) {
        ColumnFamilyHandle featureTypeHandle = this.DefaultKVStore.getColumnFamilyHandle(CF_FSCHEMA);
        try {
            byte[] value = this.DefaultKVStore.getDb().get(featureTypeHandle, id.getBytes());
            return Mono.just(KryoUtils.deserializeObject(value, TileSet.class));
        } catch (RocksDBException e) {
            throw new RocksIOException("find TileSet by " + id);
        }
    }

    /**
     * 写入瓦片字节片段
     * @param tileset
     * @param segment
     */
    public void putTileSegment(TileSet tileset,Map<String,byte[]> segment) {
        String uid = tileset.getUid();
        TileSetType type = tileset.getType();
        String schemaUrl = tileset.getSchemaUrl();
        String metadataUrl = tileset.getMetadataUrl();
        String schemaPath = DirUtils.getTileRootPath(uid, schemaUrl);
        String metadataPath = DirUtils.getTileRootPath(uid, metadataUrl);
        //TODO TileSet保持集群最终一致性，记录VNode, SyncSeq, Key到数据库
        try {
            //1->tileset
            tileset.setSchemaUrl(schemaPath);
            tileset.setMetadataUrl(metadataPath);
            byte[] tileset_value = KryoUtils.serializeObject(tileset);
            this.DefaultKVStore.getDb().put(this.DefaultKVStore.getColumnFamilyHandle(CF_FSCHEMA), uid.getBytes(), tileset_value);
            //2->schema ->layer.json or tileset.json
            byte[] schema_key = schemaPath.getBytes();
            byte[] schema_value = Files.readAllBytes(Paths.get(schemaUrl));
            this.DefaultKVStore.getDb().put(this.DefaultKVStore.getColumnFamilyHandle(CF_FSCHEMA), schema_key, schema_value);
            //3->if 3dtile put scene.json
            if (type.equals(TileSetType.M3DTILE)) {
                byte[] meta_key = metadataPath.getBytes();
                byte[] meta_value = Files.readAllBytes(Paths.get(metadataUrl));
                this.DefaultKVStore.getDb().put(this.DefaultKVStore.getColumnFamilyHandle(CF_FMETADATA), meta_key, meta_value);
            }
        } catch (IOException e) {
            throw new LocalFileIOException("not found tile path");
        } catch (RocksDBException e) {
            throw new RocksIOException("Insert Tile for " + uid);
        }
    }
    /**
     * 写入本地瓦片数据集
     *
     * @param tileset
     */
    public void putLocalTileSet(TileSet tileset) {
        String uid = tileset.getUid();
        List<String> tiles = tileset.getTiles();
        //try insert tiles data
        List<String> fail_list = batchInsertTile(uid, tiles);
        if (fail_list.size() == 0) {
            TileSetType type = tileset.getType();
            String schemaUrl = tileset.getSchemaUrl();
            String metadataUrl = tileset.getMetadataUrl();
            String schemaPath = DirUtils.getTileRootPath(uid, schemaUrl);
            String metadataPath = DirUtils.getTileRootPath(uid, metadataUrl);
            try {
                //1->tileset
                tileset.setSchemaUrl(schemaPath);
                tileset.setMetadataUrl(metadataPath);
                byte[] tileset_value = KryoUtils.serializeObject(tileset);
                this.DefaultKVStore.getDb().put(this.DefaultKVStore.getColumnFamilyHandle(CF_FSCHEMA), uid.getBytes(), tileset_value);
                //2->schema ->layer.json or tileset.json
                byte[] schema_key = schemaPath.getBytes();
                byte[] schema_value = Files.readAllBytes(Paths.get(schemaUrl));
                this.DefaultKVStore.getDb().put(this.DefaultKVStore.getColumnFamilyHandle(CF_FSCHEMA), schema_key, schema_value);
                //3->if 3dtile put scene.json
                if (type.equals(TileSetType.M3DTILE)) {
                    byte[] meta_key = metadataPath.getBytes();
                    byte[] meta_value = Files.readAllBytes(Paths.get(metadataUrl));
                    this.DefaultKVStore.getDb().put(this.DefaultKVStore.getColumnFamilyHandle(CF_FMETADATA), meta_key, meta_value);
                }
            } catch (IOException e) {
                throw new LocalFileIOException("not found tile path");
            } catch (RocksDBException e) {
                throw new RocksIOException("Insert Tile for " + uid);
            }
        } else {
            //try
            try {
                Thread.sleep(1000);
                tileset.setTiles(fail_list);
                putLocalTileSet(tileset);
            } catch (InterruptedException e) {
                logger.error(e.getMessage());
            }
        }
    }

    /**
     * batch Insert Tile
     *
     * @param uid
     * @param tiles
     * @return
     */
    public List<String> batchInsertTile(String uid, List<String> tiles) {
        List<String> fail_list = new ArrayList<>();
        //batch write tile
        Partitions.manyToOne(tiles, MAX_BATCH_WRITE_SIZE, (Function<List<String>, Long>) segment -> {
            try (final WriteBatch batch = new WriteBatch()) {
                for (String url : segment) {
                    String path = DirUtils.getTilePath(uid, url);
                    if (path != null) {
                        byte[] key = path.getBytes();
                        byte[] value = Files.readAllBytes(Paths.get(url));
                        batch.merge(key, value);//defaultHandle
                    } else {
                        fail_list.add(url);
                    }
                }
                this.DefaultKVStore.getDb().write(this.DefaultKVStore.getWriteOptions(), batch);//defaultHandle
                //TODO 写入记录点
                long lastSeq = this.DefaultKVStore.getDb().getLatestSequenceNumber();
                System.out.println(lastSeq);
                return batch.getDataSize();
            } catch (IOException e) {
                throw new LocalFileIOException("not found tile path");
            } catch (RocksDBException e) {
                throw new RocksIOException("Batch Insert Tile for " + uid);
            }
        });
        return fail_list;
    }


}
